package com.emc.elc.rabbitmq.wapper;

import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

import com.emc.elc.rabbitmq.wapper.Handler.MessageHandler;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

public class QueueConsumerThread extends Thread implements Serializable{
    private static final long serialVersionUID = 1L;
    
    private QueueingConsumer consumer;
    private Map<String, MessageHandler> messageHandlers = new HashMap<String, MessageHandler>();
    private Channel channel;
    private String queueId;
    private boolean isRunning;
    private String topic;
    private String exchange;
    
    public QueueConsumerThread(Channel channel, String queue, String topic, MessageHandler messageHandler, String exchange) throws IOException {
        this.messageHandlers.put(messageHandler.getClass().getName(), messageHandler);
        this.topic = topic;
        this.exchange = exchange;
        this.queueId = queue;

        if (null == topic) {
            channel.exchangeDeclare(exchange, "direct");
            channel.queueBind(queue, exchange, queue);
        } else {
            channel.exchangeDeclare(exchange, "topic");
            channel.queueBind(queueId, exchange, topic);
        }

        this.channel = channel;

        this.consumer = new QueueingConsumer(channel);
        this.channel.basicConsume(queueId, true, consumer);

        this.start();
    }

    public void stopListning() {
        if (!this.isRunning()) {
            return;
        }
        try {
            wait(0);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void startListning() {
        if (!this.isInterrupted()) {
            return;
        }
        notify();
    }

    public QueueConsumerThread addMessageHandlers(MessageHandler messageHandler) {
        this.messageHandlers.put(messageHandler.getClass().getName(), messageHandler);
        return this;
    }
    
    public QueueConsumerThread removeMessageHandlers(MessageHandler messageHandler) {
        this.messageHandlers.remove(messageHandler.getClass().getName());
        return this;
    }

    @SuppressWarnings("deprecation")
    public void destroyConsumer() {
        isRunning = false;
        this.stop();

        try {
            this.channel.exchangeUnbind(queueId, exchange, topic);
            this.channel.queueDelete(queueId);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean isRunning() {
        isRunning = Thread.currentThread().isAlive() && Thread.currentThread().isInterrupted();
        return isRunning;
    }
    
    public String getQueueId() {
        return queueId;
    }

    public void run() {
        while (true) {
            try {
                QueueingConsumer.Delivery delivery = this.consumer.nextDelivery();
                if (null != delivery) {
                    String content = new String(delivery.getBody());
                    for (MessageHandler messageHandler : messageHandlers.values()) {
                        messageHandler.handle(content, delivery.getEnvelope().getRoutingKey(), delivery.getProperties());
                    }
                }
            } catch (ShutdownSignalException e) {
                e.printStackTrace();
            } catch (ConsumerCancelledException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
